package com.liuyu.common.framework.event.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.liuyu.common.framework.rocketmq.properties.RocketMqConsumeProperties;
import com.liuyu.common.framework.queue.model.QueueMessageWrapper;
import com.liuyu.common.framework.rocketmq.status.context.MessageQueueProcessStatusContext;
import com.liuyu.common.framework.rocketmq.vo.MqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * rocket 消息消费
 * MessageListenerConcurrently 并行，多个线程消费，不能保证顺序消费。
 * MessageListenerOrderly 顺序消费，在发送的时候也需要保证发送在同一个队列中。
 *
 * @author: liuyu
 * @date: 2022/4/10 18:44
 * @version:
 */

@Component
@Slf4j
@RefreshScope
public class RocketOrderlyMessageListener implements MessageListenerOrderly {
    private static final String SW_8 = "sw8";

    /**
     * 消费严格模式，true
     */
    @Autowired
    private RocketMqConsumeProperties rocketMqConsumeProperties;
    @Autowired
    private MessageQueueProcessStatusContext messageQueueProcessStatusContext;

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
        String thread = Thread.currentThread().getName();
        int size = list.size();
        long start = System.currentTimeMillis();
        log.info("顺序消息 {}接收到消息数:{}", thread, size);
        for (MessageExt messageExt : list) {
            QueueMessageWrapper<MessageExt> queueMessageWrapper = new QueueMessageWrapper(messageExt.getTopic(), messageExt.getTags(), messageExt);
            //主要解決tag不正常的情况。
            String message = new String(messageExt.getBody());
            MqMessage taskMqMessage = JSON.parseObject(message, new TypeReference<MqMessage>() {
            });
            queueMessageWrapper.setTag(taskMqMessage.getTags());
            try {
                messageQueueProcessStatusContext.processQueueMessage(queueMessageWrapper);
            } catch (Exception e) {
                if (log.isDebugEnabled()) {
                    log.error("[顺序消息处理消息失败][{}]：{}", messageExt.getMsgId(), e.getMessage(), e);
                } else {
                    log.error("[顺序消息处理消息失败][{}]：{}", messageExt.getMsgId(), e.getMessage());
                }
                if (!rocketMqConsumeProperties.isSkipErrors()) {
                    //判断是否是严格模式，如果为true，则通过ack 重试，否则保存数据库通过数据库来确定
                    if (rocketMqConsumeProperties.getStrict(queueMessageWrapper.getTopic(), queueMessageWrapper.getTag())) {
                        log.error("[顺序消息处理消息失败][{}]：策略:重试", messageExt.getMsgId(), e.getMessage());
                        //延迟
                        consumeOrderlyContext.setSuspendCurrentQueueTimeMillis(rocketMqConsumeProperties.getSuspendCurrentQueueTimeMillis());
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                    //这个必须成功，否则会导致消息丢失，当这个没有成功，则直接结束，可能会导致重复消费问题。
                    try {
                        if (!messageQueueProcessStatusContext.saveFailMessage(queueMessageWrapper, e)) {
                            log.error("[顺序消息处理失败后入库失败][{}]：{}", messageExt.getMsgId(), e.getMessage(), e);
                            //如果保存失败，则稍后重发   SUSPEND_CURRENT_QUEUE_A_MOMENT
                            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                        }
                    } catch (Exception exception) {
                        log.error("[顺序消息处理失败后入库失败][{}]：{}", messageExt.getMsgId(), exception.getMessage(), exception);
                        //如果保存失败，则稍后重发
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
            }
        }
        log.info("顺序消息{}成功处理消息数:{},耗时:{}", thread, size, System.currentTimeMillis() - start);
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
